package com.gxljc.bear.crawler.related;

import java.sql.SQLException;

import com.gxljc.bear.crawler.base.DolphinCrawlerConf;
import com.gxljc.bear.crawler.base.DolphinCrawlerConsts;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;

import com.gxljc.commons.util.sql.BaseDao;
import com.gxljc.commons.util.sql.BaseDaoFactory;

import java.io.File;
import java.io.InputStreamReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.BufferedReader;

/**
 * 相关电影和重名电影写入MySQL
 * 
 * @author yunbiao,yihao
 * @since 2015-6-11 上午10:25:40
 */
public class RelatedMovie {
	public final static Logger LOG = Logger.getLogger(RelatedMovie.class);

	// write the series movies and same name movies in t_movie_cluster
	public void WriteInDB_cluster(int cluster_id, int type, String title,
			BaseDao baseDao) {
		String sql;
		sql = String.format(
				"INSERT INTO %s (cluster_id,type,title) VALUE (%d,%d,\"%s\")",
				DolphinCrawlerConsts.MYSQL_TABLE_MOVIE_CLUSTER, cluster_id,
				type, title);
		try {
			baseDao.batchExecute(sql);
		} catch (SQLException e) {
			LOG.error("Insert Error!", e);
		}
	}

	// write the series movies and same name movies in t_movie_related
	public void WriteInDB_related(int cluster_id, String subject_id,
			String title, BaseDao baseDao) {
		String sql;
		sql = String
				.format("INSERT INTO %s (cluster_id,subject_id,title) VALUE (%d,\"%s\",\"%s\")",
						DolphinCrawlerConsts.MYSQL_TABLE_MOVIE_RELATED,
						cluster_id, subject_id, title);
		try {
			baseDao.batchExecute(sql);
		} catch (SQLException e) {
			LOG.error("Insert Error!", e);
		}
	}

	// judge whether the string has more than two ###
	private boolean is_series(String str) {
		String[] abc = str.split("###");
		if (abc.length > 3)
			return true;
		else
			return false;
	}

	// judge whether the string has more than one ###
	private boolean is_same(String str) {
		String[] abc = str.split("###");
		if (abc.length > 2)
			return true;
		else
			return false;
	}

	public void Excute() {
		File SeriesFile = new File("IDforSameSeries.txt");
		File SameNameFile = new File("IDforSameName.txt");
		// configuration
		Configuration conf = DolphinCrawlerConf.getInstance();
		String dbString = conf.get(DolphinCrawlerConsts.PARAM_MYSQL_URL);
		BaseDao baseDao = BaseDaoFactory.getDaoBaseInstance(dbString);
		try {
			InputStreamReader frSeries = new InputStreamReader(
					new FileInputStream(SeriesFile), "utf-8");
			InputStreamReader frSame = new InputStreamReader(
					new FileInputStream(SameNameFile), "utf-8");
			BufferedReader brSeries = new BufferedReader(frSeries);
			BufferedReader brSame = new BufferedReader(frSame);
			String line = brSeries.readLine();
			String[] abc;
			int cluster_id = 1;
			int len;
			// for Same Series movies;
			while (line != null) {
				if (is_series(line)) {
					abc = line.split("###");
					len = abc.length;
					WriteInDB_cluster(cluster_id, 1, abc[len - 1], baseDao);
					for (int i = 0; i < len - 2; i++) {
						WriteInDB_related(cluster_id, abc[i], abc[i + 1],
								baseDao);
						i++;
					}
					cluster_id = cluster_id + 1;
				}
				line = brSeries.readLine();
			}
			// for Same Name movies;
			line = brSame.readLine();
			while (line != null) {
				if (is_same(line)) {
					abc = line.split("###");
					len = abc.length;
					WriteInDB_cluster(cluster_id, 2, abc[len - 1], baseDao);
					for (int i = 0; i < len - 1; i++) {
						WriteInDB_related(cluster_id, abc[i], abc[len - 1],
								baseDao);
					}
					cluster_id = cluster_id + 1;
				}
				line = brSame.readLine();
			}
			brSeries.close();
			brSame.close();
		} catch (IOException e) {
			System.out.println("IOException" + e);
		}
	}
}
